package com.eagle.scali.contentcenter.listener.rocketmq;

import com.alibaba.fastjson.JSON;

import com.eagle.scali.contentcenter.dao.messaging.RocketmqTransactionLogMapper;
import com.eagle.scali.contentcenter.domain.dto.content.ShareAuditDTO;
import com.eagle.scali.contentcenter.domain.entity.messaging.RocketmqTransactionLog;
import com.eagle.scali.contentcenter.service.content.ShareService;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

/**
 * 测试事务
 * 1 断点在executeLocalTransaction的 return RocketMQLocalTransactionState.COMMIT;
 * 	 checkLocalTransaction的 MessageHeaders headers = msg.getHeaders();
 * 2 执行到断点时候，可以看到content_center的数据更新了，但user_center的没有更新
 * 3 杀进程
 *   jps
 *   taskkill -PID 10404 -F
 * 4 重新启动ContentCenterApplication
 * 5 自动执行checkLocalTransaction中的断点，更新user_center数据
 * 6 事务更新完毕
 */
@RocketMQTransactionListener(txProducerGroup = "tx-add-bonus-group")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class AddBonusTransactionListener implements RocketMQLocalTransactionListener {
    private final ShareService shareService;
    private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();

        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Integer shareId = Integer.valueOf((String) headers.get("share_id"));
        String dtoString = (String) headers.get("dto");
        ShareAuditDTO auditDTO = JSON.parseObject(dtoString, ShareAuditDTO.class);

        try{
        	//正式提交生产者
			this.shareService.auditByIdWithRocketMqLog(shareId, auditDTO, transactionId);
			return RocketMQLocalTransactionState.COMMIT;
		}catch (Exception e){
//        	当auditByIdInDB出错的时候，会执行回滚
			return RocketMQLocalTransactionState.ROLLBACK;
		}
    }

	/**
	 * 服务端没有收到提交信息，会回查生产者，例如return RocketMQLocalTransactionState.COMMIT;时候网络卡了
	 * 生产者在这里检查本地事务的状态
	 * @param msg
	 * @return
	 */
	@Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);

        // select * from xxx where transaction_id = xxx
        RocketmqTransactionLog transactionLog = this.rocketmqTransactionLogMapper.selectOne(
            RocketmqTransactionLog.builder()
                .transactionId(transactionId)
                .build()
        );
        //找到记录
        if (transactionLog != null) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}
